package com.hex.ds.hdtp.core.app.data.transfer.handler.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateTime;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.hex.bigdata.hbda.utils.StringUtils;
import com.hex.ds.hdtp.core.app.common.entity.BasicContrastInfo;
import com.hex.ds.hdtp.core.app.common.util.JaveShellUtil;
import com.hex.ds.hdtp.core.app.data.contrast.service.IContrastTableStructureService;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferLogConverter;
import com.hex.ds.hdtp.core.app.data.transfer.converter.IDataTransferSqlStructureConverter;
import com.hex.ds.hdtp.core.app.data.transfer.dto.DataTransferSqlStructureDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.request.DataTransferLogRequestDto;
import com.hex.ds.hdtp.core.app.data.transfer.dto.response.DataTransferListResponseDto;
import com.hex.ds.hdtp.core.app.data.transfer.handler.IDataTransferHandlerFileService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferListService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferLogService;
import com.hex.ds.hdtp.core.app.data.transfer.service.IDataTransferTaskService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceConfInfoService;
import com.hex.ds.hdtp.core.app.metadata.service.ISourceInfoService;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcConnectUtil;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcUtil;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.repository.impl.SourceConfInfoRepository;
import com.jcraft.jsch.Session;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.postgresql.core.BaseConnection;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.io.UncheckedIOException;
import java.nio.charset.Charset;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.nio.file.StandardOpenOption;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

@Slf4j
@Service
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public abstract class AbstractDataTransferHandlerFileService implements IDataTransferHandlerFileService {

    private final ISourceInfoService sourceInfoService;

    private final ISourceConfInfoService sourceConfInfoService;

    private final IDataTransferLogService dataTransferLogService;

    private final IDataTransferListService dataTransferListService;

    private final IDataTransferTaskService dataTransferTaskService;

    private final IDataTransferLogConverter dataTransferLogConverter;

    private final IDataTransferSqlStructureConverter dataTransferSqlStructureConverter;

    private final IContrastTableStructureService contrastTableStructureService;

    protected final JaveShellUtil javeShellUtil;

    private final SourceConfInfoRepository sourceConfInfoRepository;

    private static final String DELIMITER = "Separator";

    private static final String ENCODE = "Encode";

    private static final String FILE_ENCODE = "FileEncode";

    private static final String USER = "serverUserName";

    private static final String IP = "Ip";

    private static final String PWD = "serverUserPwd";

    private static final String PATH = "Path";

    private final static int REMOTEPORT = 22;

    private static final String SHELL_RESULT = "SHELL_RESULT";

    private static final String SHELL_ERR_RESULT = "SHELL_ERR_RESULT";

    private static final String ANALYSIS_RESULT = "ANALYSIS_RESULT";

    @Value("${data.transfer.file-save-path}")
    public String localTdPath;

    /**
     * @Method dataJdbcTransfer
     * @param dataTransferSqlStructureDto
     * @param isDataAnalysis
     * @Return void
     * @Description 数据迁移执行
     * @Author Yin.Yang
     * @Date 2023/10/17 19:23
     * @Version V1.0
     */
    @Override
    @Async("dataTransferAsyncTaskExecutor")
    public void dataFileTransfer(DataTransferSqlStructureDto dataTransferSqlStructureDto, boolean isDataAnalysis) {
        boolean dataTransferSuccess = true;
        String dataAnalysisResult = dataAnalysis(dataTransferSqlStructureDto, isDataAnalysis);
        for (DateTime dateTime : dataTransferSqlStructureDto.getBizDates()) {
            String bizDate = DateUtil.format(dateTime, "yyyyMMdd");
            String dataTransferLogId = addDataTransferLog(dataTransferSqlStructureDto, bizDate, dataAnalysisResult, isDataAnalysis);
            if (!isDataAnalysis || StrUtil.isBlank(dataAnalysisResult)) {
                dataTransferSuccess = doDataFileTransfer(dataTransferLogId, bizDate, dataTransferSqlStructureDto);
            } else {
                dataTransferSuccess = false;
            }
        }
        updateDataTransferTask(dataTransferSqlStructureDto, dataTransferSuccess);
    }

    /**
     * @Method doDataJdbcTransfer
     * @param dataTransferLogId
     * @param bizDate
     * @param dataTransferSqlStructureDto
     * @Return boolean
     * @Description 执行数据迁移
     * @Author Yin.Yang
     * @Date 2023/10/17 19:23
     * @Version V1.0
     */
    protected boolean doDataFileTransfer(String dataTransferLogId, String bizDate, DataTransferSqlStructureDto dataTransferSqlStructureDto) {
        Map resultMap = new HashMap();
        Map<String, String> sourceConfMap = dataTransferSqlStructureDto.getSourceConfMap();
        Map<String, String> targetConfMap = dataTransferSqlStructureDto.getTargetConfMap();
        String delimiter = getDelimiter(sourceConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType().toLowerCase() + DELIMITER));
        String sourceEncoding = sourceConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType().toLowerCase() + FILE_ENCODE);
        String targetEncoding = targetConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getTargetConfList()).getConfType().toLowerCase() + ENCODE);
        String sourceUser = sourceConfMap.get(USER);
        String sourceIp = sourceConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType().toLowerCase() + IP);
        String sourcePwd = sourceConfMap.get(PWD);
        String filePath = sourceConfMap.get(CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType().toLowerCase() + PATH) + dataTransferSqlStructureDto.getSourceTable() + ".txt";
        String tempPath = localTdPath + CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType().toLowerCase()+ "/" + dataTransferSqlStructureDto.getSourceTable() + "_" +dataTransferSqlStructureDto.getDataTransferTaskPkId() + ".txt";
        String targetTableName = dataTransferSqlStructureDto.getTargetTableName();
        try {
            // 建立目标数据库连接
            SourceInfoPo targetSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getTargetConnPkId());
            List<SourceConfInfoPo> targetConfInfoPoList = sourceConfInfoService.queryListBySourceCode(targetSource.getSourceCode());
            Connection jdbcConnection = JdbcUtil.getJdbcConnection(targetConfInfoPoList);
            BaseConnection baseConn = jdbcConnection.unwrap(BaseConnection.class);
            // 检测本地目录是否存在，并取出数据库中数据的文件
            checkLocalDir(CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType().toLowerCase());
            log.info("文本载数-连接数据文件服务器，IP：" + sourceIp + " User:" + sourceUser);
            Session session = javeShellUtil.getConnectSession(sourceUser, sourceIp, sourcePwd, REMOTEPORT);
            log.info("文本载数-将数据文件移动至临时文件，filePath：" + filePath + " tempPath:" + tempPath);
            javeShellUtil.pullFileFromConnect(session, filePath , tempPath);
            // 将文件中的数据导入到目标表中
            String tempFilePath = addFilePartition(tempPath, bizDate, delimiter, sourceEncoding, targetEncoding, resultMap);
            copyFromFile(baseConn, tempFilePath, targetTableName, delimiter, targetEncoding, resultMap);
            resultMap.put(SHELL_RESULT, "0");
            resultMap.put(SHELL_ERR_RESULT,"迁移成功");
            resultMap.put(ANALYSIS_RESULT, getResultCount(dataTransferSqlStructureDto));
        } catch (Exception e) {
            log.error("【{}】数据同步失败: {}", dataTransferSqlStructureDto.getDataTransferSqlFileName(), e.getMessage(), e);
            resultMap.put(SHELL_RESULT, "1");
            resultMap.put(ANALYSIS_RESULT, "迁移失败,无法汇总");
            if (resultMap.get(SHELL_ERR_RESULT) == null) {
                resultMap.put(SHELL_ERR_RESULT, e.getMessage());
            }
        } finally {
            updateDataTransferLog(dataTransferLogId, resultMap);
        }
        return "0".equals(resultMap.get(JaveShellUtil.SHELL_RESULT));
    }

    /**
     * @Method updateDataTransferLog
     * @param dataTransferLogPkId
     * @param resultMap
     * @Return void
     * @Description 更新数据迁移记录结果
     * @Author Yin.Yang
     * @Date 2023/10/17 19:07
     * @Version V1.0
     */
    protected void updateDataTransferLog(String dataTransferLogPkId, Map<String, String> resultMap) {
        DataTransferLogRequestDto dataTransferLogRequestDto = new DataTransferLogRequestDto();
        dataTransferLogRequestDto.setPkId(dataTransferLogPkId).setEndTime(DateUtil.now());
        dataTransferLogRequestDto.setStatus("0".equals(resultMap.get(JaveShellUtil.SHELL_RESULT)) ? "SUCCESS" : "FAIL");
        dataTransferLogRequestDto.setAnalysisResult(resultMap.get(ANALYSIS_RESULT));
        String errorMessage = "转换执行SQL: " + resultMap.get("doTransferSql") + "\n" +  "迁移信息: " + resultMap.get(JaveShellUtil.SHELL_ERR_RESULT);
        dataTransferLogRequestDto.setErrorMessage(errorMessage);
        dataTransferLogService.modifyById(dataTransferLogRequestDto);
    }

    /**
     * @Method updateDataTransferTask
     * @param dataTransferSqlStructureDto
     * @param dataTransferSuccess
     * @Return void
     * @Description 更新数据迁移任务成功、失败数量和状态
     * @Author Yin.Yang
     * @Date 2023/10/17 19:10
     * @Version V1.0
     */
    private void updateDataTransferTask(DataTransferSqlStructureDto dataTransferSqlStructureDto, boolean dataTransferSuccess) {
        if (dataTransferSuccess) {
            dataTransferTaskService.modifySuccessNum(dataTransferSqlStructureDto.getDataTransferTaskPkId());
        } else {
            dataTransferTaskService.modifyFailNum(dataTransferSqlStructureDto.getDataTransferTaskPkId());
        }
        dataTransferTaskService.modifyStatus(dataTransferSqlStructureDto.getDataTransferTaskPkId());
    }

    /**
     * @Method dataAnalysis
     * @param dataTransferSqlStructureDto
     * @param isDataAnalysis
     * @Return java.lang.String
     * @Description 数据迁移前数据分析（源表和目标表是否存在、源表和目标表字段对比、源表是否为空）
     * @Author Yin.Yang
     * @Date 2023/10/17 19:11
     * @Version V1.0
     */
    private String dataAnalysis(DataTransferSqlStructureDto dataTransferSqlStructureDto, boolean isDataAnalysis) {
        if (isDataAnalysis) {
            SourceInfoPo sourceSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getSourceConnPkId());
            SourceInfoPo targetSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getTargetConnPkId());
            BasicContrastInfo basicContrastInfo = dataTransferSqlStructureConverter
                    .toBasicContrastInfo(dataTransferSqlStructureDto, sourceSource, targetSource);
            if(!contrastTableStructureService.contrastCreateTable(basicContrastInfo).getResult()) {
                return "源表或目标表不存在";
            } else if (!contrastTableStructureService.contrastFieldNameWithMap(basicContrastInfo, dataTransferSqlStructureDto.getDataTransferListPkId()).getResult()) {
                return "源表和目标表字段映射存在问题";
            }
        }
        return null;
    }


    /**
     * @Method addDataTransferLog
     * @param dataTransferSqlStructureDto
     * @param bizDate
     * @param dataAnalysisResult
     * @param isDataAnalysis
     * @Return java.lang.String
     * @Description 新增数据迁移记录
     * @Author Yin.Yang
     * @Date 2023/10/17 19:12
     * @Version V1.0
     */
    private String addDataTransferLog(DataTransferSqlStructureDto dataTransferSqlStructureDto, String bizDate,
                                      String dataAnalysisResult, boolean isDataAnalysis) {
        DataTransferListResponseDto dataTransferList = dataTransferListService.queryById(dataTransferSqlStructureDto.getDataTransferListPkId());
        DataTransferLogRequestDto dataTransferLog = dataTransferLogConverter.toDto(dataTransferList, bizDate, dataAnalysisResult);
        if (isDataAnalysis && StrUtil.isNotBlank(dataAnalysisResult)) {
            dataTransferLog.setStatus("FAIL").setEndTime(DateUtil.now());
        }
        return dataTransferLogService.add(dataTransferLog);
    }

    /**
     * @Method addFilePartition
     * @Param tempPath
     * @Param bizDate
     * @Param delimiter
     * @Param sourceEncoding
     * @Param targerEncoding
     * @Return java.lang.String
     * @Description 向源数据文件拼接分区字段
     * @Author Yin.Yang
     * @Date 2023/10/18 15:48
     * @Version V2.0
     */
    private String addFilePartition(String tempPath, String bizDate, String delimiter, String sourceEncoding, String targerEncoding, Map resultMap) throws IOException {
        Path path = Paths.get(tempPath);
        Charset sourceCharset = Charset.forName(sourceEncoding);
        Charset targetCharset = Charset.forName(targerEncoding);
        try(Stream<String> input = Files.lines(path, sourceCharset)) {
            List<String> updatedLines = input
                    .map(line -> line + delimiter + bizDate)
                    .collect(Collectors.toList());
            Files.write(path, updatedLines, targetCharset, StandardOpenOption.TRUNCATE_EXISTING, StandardOpenOption.CREATE);
        } catch (UncheckedIOException e) {
            String logMsg = "文本载数-数据文件编码格式有误：" + sourceCharset;
            log.info(logMsg);
            resultMap.put(SHELL_RESULT, "1");
            resultMap.put(SHELL_ERR_RESULT, logMsg);
            return null;
        }
        return path.toString();
    }



    /**
     * @Method checkLocalDir
     * @Return void
     * @Description 检查本地目录是否存在，不存在则创建
     * @Author Yin.Yang
     * @Date 2023/10/19 18:15
     * @Version V1.0
     */
    private void checkLocalDir(String sourceType) {
        Path path = Paths.get(localTdPath + sourceType + "/");
        if (!Files.exists(path)) {
            try {
                Files.createDirectories(path);
            } catch (IOException e) {
                log.error("本地目录存储地址错误：{}", e.getMessage(), e);
            }
        }
    }

    /**
     * @Method getDelimiter
     * @param delimiter
     * @Return java.lang.String
     * @Description 获取正确的分隔符
     * @Author Yin.Yang
     * @Date 2023/10/25 15:40
     * @Version V1.0
     */
    private String getDelimiter(String delimiter) {
        // 如果分隔符符合\00X则转为ascii码，否则直接使用
        if (delimiter.startsWith("\\") && delimiter.length() > 1) {
            int asciiCode = Integer.parseInt(delimiter.substring(3));
            delimiter = Character.toString((char) asciiCode);
        }
        return delimiter;
    }

    /**
     * @Method getResultCount
     * @Param dataTransferSqlStructureDto
     * @Return java.lang.String
     * @Description 数据汇总情况
     * @Author xin.zhao
     * @Date 2023/10/26 10:42
     * @Version V1.0
     */
    protected String getResultCount(DataTransferSqlStructureDto dataTransferSqlStructureDto) {
        SourceInfoPo targetSource = sourceInfoService.queryByPkId(dataTransferSqlStructureDto.getTargetConnPkId());
        List<SourceConfInfoPo> targetConfInfoList = sourceConfInfoRepository.queryListBySourceCode(targetSource.getSourceCode());
        String sql = "SELECT COUNT(*) FROM %s.%s";
        String targetSql = String.format(sql, dataTransferSqlStructureDto.getTargetDatabase(), dataTransferSqlStructureDto.getTargetTable());
        if (StringUtils.isNotBlank(dataTransferSqlStructureDto.getWhereCondition())) {
            targetSql = targetSql + " WHERE " + dataTransferSqlStructureDto.getWhereCondition();
        }
        Long targetLongResult = null;
        try {
            targetLongResult = JdbcConnectUtil.getLongResult(targetConfInfoList, targetSql);
        } catch (SQLException e) {
            Throwable rootCause = findRootCause(e);
            if (rootCause != null) {
                return rootCause.getMessage();
            }
        }
        return CollUtil.getFirst(dataTransferSqlStructureDto.getSourceConfList()).getConfType() + "数据库数据总量：文件载数方式\n" + CollUtil.getFirst(dataTransferSqlStructureDto.getTargetConfList()).getConfType() + "数据库数据总量：" + targetLongResult.toString();
    }

    /**
     * @Method findRootCause
     * @Param e
     * @Return java.lang.Throwable
     * @Description 辅助方法，递归查找异常根本原因
     * @Date 2023/10/26 10:42
     * @Version V1.0
     */
    private static Throwable findRootCause(Throwable e) {
        Throwable cause = e;
        while (cause.getCause() != null) {
            cause = cause.getCause();
        }
        return cause;
    }

    /**
     * @Method copyFromFile
     * @Param baseConn
     * @Param tempFilePath
     * @Param targetTableName
     * @Param delimiter
     * @Param targetEncoding
     * @Param resultMap
     * @Return void
     * @Description 从指定的文件中读取数据，并将其导入到指定的表中
     * @Author Yin.Yang
     * @Date 2023/10/18 16:26
     * @Version V2.0
     */
    protected abstract void copyFromFile(BaseConnection baseConn, String tempFilePath, String targetTableName, String delimiter, String targetEncoding, Map<String, String> resultMap) throws SQLException, IOException;

}
